package edu.mercer.sse554.bkp;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.NumberFormat;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.xerial.snappy.SnappyInputStream;
import org.xerial.snappy.SnappyOutputStream;

import com.google.common.base.Function;
import com.google.common.base.Optional;
import com.google.common.base.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.ByteSink;
import com.google.common.io.ByteSource;
import com.google.common.io.Files;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.SimpleTimeLimiter;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

/**
 * SnappyCompression<p>
 * 
 * Compress a File. Produce collection of results of the compression 
 * operation.
 * 
 * Bubbles IO checked exceptions unmodified: not wrapped.<p>
 * 
 * @see CheckedFuture
 */

public class SnappyCompression 

		implements 
		
		Callable<TaskResults<ImmutableMap<String,String>>>,

		CheckedFuture<TaskResults
					 		<ImmutableMap<String,String>>, BackupException>,
					 		
		ConfigurationConstants
{
	/** builder for this: with any construction properties */
	
	private Optional<SnappyCompressionBuilder> _jobBuilder = 
			Optional.absent();

	/** size of file to be compressed */
	
	private Long _filesize = 0L;

	/** percentage compression completed */
	
	private Float _compressCompleted = 0.0F;

	/** percentage un-compression completed */

	private Float _uncompressCompleted = 0.0F;
	
	/** percentage compression result of compressing file */
	
	private String _shrinkage = "";
	
	/** exception / error encountered */
	
	private Boolean _abend = false;
	
	/** if decompression, equality validation requested and completed */
	
	private Boolean _validated = false;
	
	/** compression request finished: completed, abended, timed-out,... */
	
	private Boolean _finished = false;
	
	/** user requests gentle cancellation */
	
	private Boolean _cancel = false;
	
	/** timer for compression task */
	
	private ConcurrTimer _clock = ConcurrTimer.factory();
	
	/** T if file compression not attempted: file too small, .sz extension */
	
	private Boolean _skipped = false;

	/** T if uncompress requested to original path */
	
	private Boolean _restore = false;
	
	/**
	 * Elapsed time while compressing/running in thread.<p> 
	 * Unavailable once outside thread such as from SnappyCRBuilder. 
	 */
	private String _elapsed = "";
	
	/** thread safe number formatter used for percentages */
	
	private static final ThreadLocal<NumberFormat> _numberFormatter = 
			new ThreadLocal<NumberFormat>()
			{
				@Override
				protected NumberFormat initialValue()
				{
					return NumberFormat.getPercentInstance();
					
				}//method
		
			};
	
	/** 
	 * user requests immediate cancellation: thread interruption during I/O.
	 * <p>
	 * 
	 * @see #pollInteruption
	 */
	
	private Boolean _cancelNow = false;
	
	
	SnappyCompression(final SnappyCompressionBuilder builder)
	{
		super();
		
		_jobBuilder = Optional.of(builder);
		
	}//constructor

	/**
	 * Instantiate SnappyCompression builder for properties and construction. 
	 * <p>
	 *  
	 * @return SnappyCompressionBuilder - new builder; i.e., intermediate 
	 * 		object instance
	 */
	
	static SnappyCompressionBuilder builder()
	{
		return new SnappyCompressionBuilder();
		
	}//method

	
	/** 
	 * @return String - final compression as formatted percentage 
	 * accomplished 
	 */

	public String shrinkage(){ return _shrinkage; }

	
	/** 
	 * @return Float - compression method percentage accomplished 
	 * 
	 * @see #compressFile
	 */

	public Float compressCompleted()
	{ 
		return _compressCompleted > 1F ? 1F : _compressCompleted; 
		
	}//method

	
	/** 
	 * @return Float - un-compression method percentage accomplished 
	 * 
	 * <p>
	 * Validation option uncompresses file to perform equality test, 
	 * else/otherwise unused.<p>
	 * 
	 * @see #unCompressFile
	 */
	
	public Float uncompressCompleted()
	{ 
		return _uncompressCompleted > 1F ? 1F : _uncompressCompleted; 
		
	}//method
	

	/**
	 * @return Boolean - T Exception/Error encountered.<p>
	 * 
	 * Includes exceptions caused by Task cancellations.
	 */
	
	public Boolean abended(){ return _abend; }
	
	
	/** @return Boolean - T builder validation method used */
	
	public Boolean validated(){	return _validated; }
	
	
	/** 
	 * @return Boolean - T this task ended: normally, timed out, 
	 * 		cancelled... 
	 */
	
	public Boolean finished(){ return _finished; }
	
	
	/**
	 * @return ConcurrentTimer - runtime duration for compression task
	 */
	
	public ConcurrTimer clock(){ return _clock; }
	
	
	/**
	 * @return String - Formatted time as Seconds spent compressing file, 
	 * 		running; i.e., while in thread.<p>
	 */
	
	public String elapsed(){ return _elapsed; }

	
	/**
	 * @return Boolean - T if file compression not attempted: file too small, 
	 * 		etc.
	 */
	
	public Boolean skipped(){ return _skipped; }

	
	public void restoreOK( final Boolean restore )
	{ 
		_restore = restore; 
		
	}//method
	
	
	/**
	 * Compressed filename.<p>
	 * 
	 * @param useTmp - include tmp folder as base of path String
	 * 
	 * @return  String - compressed filename and path as String
	 */
	
	public String fileUncompressed( final String... useTmp )
	{
		Path filePath = _jobBuilder.get()._path
				 				   .get();
		
		if( useTmp != null && useTmp.length > 0 )
		{
			return Settings.access()
						   .getProperty("TMP-FLDR") + 
						   
						   File.separator + 
						   
						   filePath.getFileName();
			
		}//if
		
		return filePath.toString();
		
	}//method

	
	private File fileUncompressedAsFile( final String... useTmp )
	{
		return fileUncompressedAsPath(useTmp).toFile();
		
	}//method


	private Path fileUncompressedAsPath( final String... useTmp )
	{
		return Paths.get( fileUncompressed(useTmp) );
		
	}//method

	
	/**
	 * UnCompressed filename.<p>
	 * 
	 * @param useTmp - include tmp folder as base of path String
	 * 
	 * @return  String - un-compressed filename and path as String
	 */
	
	public String fileCompressed( final String... useTmp )
	{
		String filePathStr = _jobBuilder.get()._path
										.get()
										.toString() + 
									 "." + 
									 	Settings.access()
									 			.getProperty("SNAPPY-EXT");
		
		if( useTmp != null && useTmp.length > 0 )
		{
			return Settings.access()
						   .getProperty("TMP-FLDR") + 
						   
						   File.separator + 
						   
						   Paths.get(filePathStr).getFileName();
			
		}//if
		
		return filePathStr;
		
	}//method


	private File fileCompressedAsFile( final String... useTmp )
	{
		return fileCompressedAsPath(useTmp).toFile();
		
	}//method


	private Path fileCompressedAsPath( final String... useTmp )
	{
		return Paths.get( fileCompressed(useTmp) );
		
	}//method
	
	
	/**
	 * Check thread should throw cancellation exception.<p>
	 * 
	 * May be caught, depending on particular Future get method employed.<p>
	 * 
	 * @return this
	 */
	
	private SnappyCompression pollCancellation()
	{
		if( _cancel ) throw new CancellationException(
				"file compression cancelled: " + fileUncompressed() );

		return this;
		
	}//method

	
	/**
	 * Check thread should throw interruption exception.<p>
	 * 
	 * Used during stream I/O for immediate interruption during prolonged
	 * 		operation.<p>
	 * 
	 * May be caught, depending on particular Future get method employed.<p>
	 * 
	 * @return this
	 * 
	 * @throws InterruptedException  parent thread interrupted or cancel
	 * 		method requested with mayInterruptIfRunning
	 * 
	 * @see #cancel
	 */
	
	private SnappyCompression pollInteruption() throws InterruptedException
	{
		if( Thread.interrupted() || _cancelNow )                  
		{
			Thread.currentThread().interrupt();
			
			throw new InterruptedException("compression task interrupted");
			
		}//if
		
		return this;
		
	}//method
	
	
	/* 
	 * Example of not using autocloseable/try-with-resources for paranoia, 
	 * resource leak prevention.<p>
	 * 
	 * Guava JavaDocs for openBufferedStream:<br>
	 * "The caller is responsible for ensuring that the returned stream is 
	 * closed."
	 * 
	 * <p>
	 * JavaDocs for InputStream:<br> 
	 * "...close method does nothing for inputstream."<p>
	 *  
	 * Google code below from ByteStream.copy method, which is used by copyTo 
	 * method of Snappy's convenience method.<p>
	 * 
	 * SnappyCompression method for stream I/O with Snappy uses this 
	 * approach.<p>
	 * 
		/
        * Copies all bytes from the input stream to the output stream.<br>
        * Does not close or flush either stream.<p>
        *
        * &#64;param from the input stream to read from<br>
        * &#64;param to the output stream to write to<br>
        * &#64;return the number of bytes copied<br>
        * &#64;throws IOException if an I/O error occurs<br>
		/
		 <pre>
    	public static long copy(InputStream from, OutputStream to)
        		throws IOException 
        {
      		byte[] buf = new byte[BUF_SIZE];
      		long total = 0;
      		while (true) {
        		int r = from.read(buf);
        		if (r == -1) {
          		break;
        		}
        		to.write(buf, 0, r);
        		total += r;
      		}
  		    return total;
  		}
  		</pre>
	 */

	/** 
	 * Perform file compression.<p>
	 * 
	 * Compress large file using stream I/O instead of loading byte arrays 
	 * into memory for Snappy compression routines.<p>
	 * 
	 * @throws BackupException  IO error with file or path
	 * 
	 * @throws BackupException  thread was interrupted by user, timeout
	 * 
	 * @throws BackupException  IO error copying compressed file from tmp 
	 * 	folder 
	 */

	private SnappyCompression compressFile() throws BackupException
	{
		Optional<InputStream> streamIn  = Optional.absent();

		Optional<OutputStream> streamOut = Optional.absent();
		
		Optional<SnappyOutputStream> streamOutCompress = Optional.absent();
		
		Boolean success = false;
		
		try
		{
			Long goSlow = Long.valueOf( Settings.access()
											    .getProperty("SLOW") );
			
			_clock.start();
			
			pollCancellation();
			
			/* file validation */
			
			
			/* used for demo of throwing exception to executor */
			
			if(fileUncompressed().contains("crash-this-task")) 
				throw new BackupException("intended failure");
			
			
			validateFile( fileUncompressedAsPath() );
			
			if( _filesize < 1 ) _filesize = fileUncompressedAsFile()
											.length();

			/* skip files already compressed by snappy */

			if( Files.getFileExtension( fileUncompressed() )
					 .equals( Settings.access()
				 	    			  .getProperty("SNAPPY-EXT") ) )
			{
				_skipped = true;
				return this;
				
			}//if
			
			/* skip small files less than buffer size */
			
			if( _filesize < Long.valueOf( 
					Settings.access().getProperty("SNAPPY-BUFF") ) )
			{
				_skipped = true;
				return this;
				
			}//if
			
			/* skip files beyond maximum allowed size */
			
			if( _filesize > Long.valueOf( 
					Settings.access().getProperty("MAX-FILE") ) )
			{
				_skipped = true;
				return this;
				
			}//if
			
			
			/* input file connection */
			
			final ByteSource rawFile = Files.asByteSource( 
					
					new File( fileUncompressedAsPath().toUri()) );  

			/* streamIn is open stream */
			
			streamIn = Optional.of( rawFile.openBufferedStream() );

			/* output file connection */
			
			fileCompressedAsFile("use tmp folder").delete();
			
			final ByteSink tmpFile = Files.asByteSink( 
					fileCompressedAsFile("use tmp folder") );

			/* streamOut is open stream */

			streamOut = Optional.of( tmpFile.openBufferedStream() );

			/* streamOutCompress is open stream */

		    streamOutCompress = Optional.of( new SnappyOutputStream(
		    		streamOut.get()) );
			
      		final byte[] buffer = new byte[ Integer
      		                                .parseInt(       
      		                           	Settings
      		                               	.access()
      		                           		.getProperty("SNAPPY-BUFF") ) ];

      		final Long maxSize = Long.valueOf( Settings
      											.access()
      											.getProperty("MAX-FILE") ); 
      		
      		Long totalRead = 0L;
      		
			pollCancellation();
			
			Float completeIntermediate = 0F;
			
      		do{
      			pollInteruption();
      			
        		Integer amountRead = streamIn.get().read(buffer);
        		if( amountRead == -1 ) break;
        		
        		streamOutCompress.get().write(buffer, 0, amountRead);
        		
        		totalRead += amountRead;

        		completeIntermediate = (totalRead) / _filesize.floatValue();
        		
       			_compressCompleted = completeIntermediate;

       			if( goSlow > 0 ) Thread.sleep( goSlow );
       			
      		}while (totalRead < maxSize);   

      		pollCancellation();
			
      		/* optional un-compression equality test */
      		
      		if( _jobBuilder.get()._validate )
      		{
    			closeStream( streamOutCompress ).
    			closeStream( streamOut ).
    			closeStream( streamIn );
      			
      			unCompressFile("use tmp folder");
      		
      			waitForFile( fileUncompressed("use tmp folder") );

      			checkArgument( Files.equal( fileUncompressedAsFile(),
      										fileUncompressedAsFile("tmp") ),
      										
      					"compression validation failure" );
      				
      			_validated = true;

    			_elapsed = _clock.stop()
   					 			 .elapsedAsString();
    			
      		}
      		else
      		{	
    			_elapsed = _clock.stop()
   					 			 .elapsedAsString();
    			
      		}//if

      		success = true;
      		
		}
		catch(final IOException          |
					InterruptedException |
					CancellationException err )
		{
			_abend = true;
			
			throw new BackupException(err);
			
		}
		catch(final Throwable err)
		{
			_abend = true;
			
			Throwables.propagateIfPossible(err, CancellationException.class);
			
			throw new BackupUncheckedException(err);
			
		}
		finally
		{
			_finished = true;

			/* chain in case one closure fails, die at that point
			 *  
			 * order of closure critical to prevent data is lost in 
			 * uncompression 
			 */
			
			closeStream( streamOutCompress ).
			closeStream( streamOut ).
			closeStream( streamIn );
			
			/* copy compressed file to original/source file path */
			
			if(success)
			{ 			
				try
				{
					waitForFile( fileCompressed("use tmp folder") );

					Files.copy( fileCompressedAsFile("use tmp"),
							
								fileCompressedAsFile() );

					fileCompressedAsFile("use tmp").delete();
	
				}
				catch( final IOException|InterruptedException err )
				{
					throw new BackupException( err );
					
				}//try
				
	      		_shrinkage = _numberFormatter.get()
	      									 .format(
	      				
	      				        (  _filesize.floatValue() - 
	      						   
	      						   new Float( fileCompressedAsFile()
	      								   	  .length() ) 
      						    ) 
      						    
      						    / new Float( _filesize )
      						    
      						);

	      		_compressCompleted = 1F;
				
	      		if( Boolean.valueOf( Settings
	      							  .access()
	      							  .getProperty("REMOVE-UNCOMPRESSED") ) 
	      			&& ! _skipped 
	      			&& ! _abend )
	      		{
	      			fileUncompressedAsFile().delete();
	      			
	      		}//if
	      		
			}//if

		}//try
		
		return this;
		
	}//method


	private void waitForFile( final String fileName ) throws
		InterruptedException
	{
		waitForFile( Paths.get(fileName) );
	
	}//method
	
	
	private void waitForFile( final Path filePath ) throws
		InterruptedException
	{
		waitForFile( filePath.toFile() );
		
	}//method
	
	
	private void waitForFile( final File fileResource ) throws 
		InterruptedException
	{
		if( ! fileResource.exists() )
		{
			LogCustom.obtainLog( Settings
					 .access()
					 .getProperty("LOG-FILE") )
					 .warning("waiting for file to exist: " 
							 + fileResource.getName() );
			
			do
			{
				pollInteruption();
				
				Thread.sleep(10);

			}while( ! fileResource.exists() );
		
		}//if
		
	}//method
	
	
	/**
	 * Close a generic stream resource.<p>
	 * 
	 * Absorb checked IOException when resource known to already be closed;
	 * i.e., when called before uncompressing file for validation.<p>
	 * 
	 * @param stream - BufferedInputStream | InputStream | SnappyInputStream 
	 * 				  | SnappyOutputStream | OutputStream
	 * 
	 * @return this
	 */
	
	private <X> SnappyCompression closeStream( final Optional<X> stream )
	{
		try
		{
			if( !stream.isPresent() )
			{
				return this;

			}//if

			if( stream.get() instanceof BufferedInputStream )
			{
				((BufferedInputStream) stream.get()).close();

			}
			else if( stream.get() instanceof InputStream )
			{
				((InputStream) stream.get()).close();

			}
			else if( stream.get() instanceof SnappyInputStream )
			{
				((SnappyInputStream) stream.get()).close();

			}
			else if( stream.get() instanceof SnappyOutputStream )
			{
				((SnappyOutputStream) stream.get()).close();

			}
			else if( stream.get() instanceof OutputStream )
			{
				((OutputStream) stream.get()).close();

			}
			else 
			{
				throw new BackupException( "stream expected " + 
						stream.get() );

			}//if
			
		}		
		catch(final IOException err)
		{
			/* 
			 * validation closes files before finally block
			 * 
			 * in this case, suppress any IO error created by attempting 
			 * again close file after already closed
			 */
			
			if( !_jobBuilder.get()._validate )
			{
				Throwables.propagate(err);
				
			}//if
			
		}
		catch(final Throwable err)
		{
			Throwables.propagate(err);
			
		}//try
		
		return this;
		
	}//method

	
	/**
	 * Perform file un-compression.<p>
	 * 
	 * Uncompresses large file using stream I/O instead of loading byte 
	 * arrays into memory for Snappy compression routines.<p>
	 * 
	 * @param useTmp - option to uncompress file to tmp folder
	 * 
	 * @return this
	 * 
	 * @throws IOException  error with file or path
	 * 
	 * @throws InterruptedException  thread was interrupted by user, timeout
	 */
	
	public SnappyCompression unCompressFile( final String... useTmp ) throws 
			IOException, InterruptedException
	{
		Optional<InputStream> streamIn  = Optional.absent();

		Optional<OutputStream> streamOut = Optional.absent();
		
		Optional<SnappyInputStream> streamInCompressed = Optional.absent();
		
		try
		{
			/* validation of compressed input file */
			
			Path filePath = fileCompressedAsPath();
			
			if( useTmp != null && useTmp.length > 0 )
			{
				filePath = fileCompressedAsPath("use tmp folder");
				
			}//if
				
			validateFile( filePath );
			
			if( _filesize < 1 ) _filesize = filePath.toFile()
													.length();

			/* input file buffered stream connection */

			final ByteSource rawFile = Files.asByteSource( 
					
					new File( filePath.toUri()) );  

			streamInCompressed = Optional.of( new SnappyInputStream(
					rawFile.openBufferedStream()) );
			
			/* output file validation, connection */

			ByteSink tmpFile;
			
			if( _restore )
			{
				/* decompress to orginal path */
				
				tmpFile = Files.asByteSink(	fileUncompressedAsFile() );
			
			}
			else
			{
				/* default use temporary folder */
				
				tmpFile = Files.asByteSink( fileUncompressedAsFile("tmp") );
				
			}//if

			streamOut = Optional.of( tmpFile.openBufferedStream() );
			
      		byte[] buffer = new byte[ Integer.parseInt( 
      									Settings
      									.access()
      									.getProperty( "SNAPPY-BUFF") ) ];

      		Long totalRead = 0L;
      		
      		final Long maxSize = Long.valueOf( Settings
      											.access()
												.getProperty("MAX-FILE") ); 
      		
      		do 
      		{
      			pollInteruption();		//thread can be halted
      			
        		Integer amountRead = streamInCompressed.get()
        											   .read(buffer);
        		if (amountRead == -1) break;
        		
        		streamOut.get()
        				 .write(buffer, 0, amountRead);
        		
        		totalRead += amountRead;
      		
        		_uncompressCompleted = (totalRead) / _filesize.floatValue();
        		
      		}while (totalRead < maxSize);
      		
      		_uncompressCompleted = 1F;
      		
		}
		catch(final IOException err)
		{
			_abend = true;
			
			throw new IOException(err);
			
		}
		catch(final InterruptedException err )
		{
			_abend = true;
			
			throw new InterruptedException();
			
		}
		catch(final Throwable err)
		{
			_abend = true;
			
			Throwables.propagate(err);
			
		}
		finally
		{
			closeStream( streamOut ).
			closeStream( streamInCompressed ).
			closeStream( streamIn );			
			
		}//try
		
		return this;
		
	}//method
	

	/**
	 * Ensure file path specified meets following requirements:
	 * 
	 * <ul>
	 * <li>not null
	 * <li>exists
	 * <li>not directory
	 * <li>readable
	 * </ul>
	 * 
	 * <p>
	 * 
	 * @param filePath - Path of file to be compressed, not null
	 * 
	 * @return this
	 * 
	 * @throws IOException  error with Path, file
	 */
	
	public SnappyCompression validateFile( final Path filePath ) throws 
			IOException
	{
		/* ensure exists */
		
		if( filePath.getFileName() == null )
			throw new IOException( "null detected in file name" );
		
		if( !filePath.toFile().exists() )
			throw new IOException( "file not found " + filePath );

		if( filePath.toFile().isDirectory() )
			throw new IOException( "file expected " + filePath );
		
		if( !filePath.toFile().canRead() )
			throw new IOException( "file unreadable " + filePath );

		return this;
		
	}//method
	

	@Override
	public void addListener(final Runnable listener, final Executor executor) 
	{
		executor.execute(listener);
		
	}//method


	@Override
	public boolean cancel(final boolean cancelNow) 
	{
		if( finished() ) return false;
		
		_cancel = true;
		
		_cancelNow = cancelNow;
		
		return true;
		
	}//method


	@Override
	public TaskResults<ImmutableMap<String,String>> get() 
		throws InterruptedException, ExecutionException 
	{
		Optional<SnappyCompressionResults> taskResults = Optional.absent();
		
		try
		{
			compressFile();

		}
		catch( final BackupException err )
		{
			if( Throwables.getRootCause(err) instanceof 
					InterruptedException )
			{
				throw new InterruptedException( Throwables.getRootCause(err)
														  .getMessage() );
			}
			else
			{
				throw new ExecutionException( Throwables.getRootCause(err) );
				
			}//if
			
		}
		finally
		{
			_finished = true;
			
			taskResults = Optional.of( 
				   SnappyCRBuilder.factory()
				   				  .setTask(this)
								  .build() );
			
		}//try
		
		return taskResults.get();
		
	}//method


	@Override
	public TaskResults<ImmutableMap<String,String>> get
		
		(final long timeout, final TimeUnit unit) 
		
		throws InterruptedException, 
			   ExecutionException, 
			   TimeoutException 
	{
		Optional<SnappyCompressionResults> taskResults = Optional.absent();

		final Duration duration = new Duration(timeout, unit);

		final SimpleTimeLimiter limiter = new SimpleTimeLimiter();
		
		if( !isDone() ) 
		{
			limiter.newProxy( this, 
					CheckedFuture.class, 
					duration.interval(), 
					duration.unit() ).get();

		}//if
		
		taskResults = Optional.of( 
				   SnappyCRBuilder.factory()
				   				  .setTask(this)
								  .build() );
			
		return taskResults.get();
		
	}//method


	@Override
	public boolean isCancelled() 
	{
		return _cancel && 
			   finished() && 
			   compressCompleted() != 1F ? true : false;
		
	}//method


	@Override
	public boolean isDone() 
	{
		return finished();
		
	}//method


	@Override
	public TaskResults<ImmutableMap<String,String>> checkedGet() 
		throws BackupException 
	{
		Optional<SnappyCompressionResults> taskResults = Optional.absent();
		
		if( !isDone() ) compressFile();

		_finished = true;

		taskResults = Optional.of( 
				SnappyCRBuilder.factory()
				.setTask(this)
				.build() );
		
		return taskResults.get();
				
	}//method


	@Override
	public TaskResults<ImmutableMap<String,String>> 
	
			checkedGet(	final long timeout, final TimeUnit unit ) 
					
			throws TimeoutException, BackupException 
	{		
		final Duration duration = new Duration(timeout, unit);

		final SimpleTimeLimiter limiter = new SimpleTimeLimiter();
		
		try 
		{
			if( !isDone() ) 
			{
				limiter.newProxy( this, 
						CheckedFuture.class, 
						duration.interval(), 
						duration.unit() ).checkedGet();
			
			}//if
			
		}
		catch( final TimeoutException err )
		{
			_abend = true;
			
			throw err;
			
		}
		catch( final Exception err )
		{
			_abend = true;
		
			throw new BackupException( err );
			
		}//try
		
		return SnappyCRBuilder.factory()
							  .setTask( this )
							  .build();		
		
	}//method

	
	@Override
	public TaskResults<ImmutableMap<String,String>> call() 
			throws BackupException 
	{
		TaskResults<ImmutableMap<String,String>> results;
		
		try 
		{
			if( !isDone() ) compressFile();

		}
		catch( final BackupException err )
		{
			throw err;
			
		}
		catch( final Throwable err ) 
		{
			throw new BackupUncheckedException( 
					Throwables.getRootCause(err) );
		
		}
		finally
		{
			results = SnappyCRBuilder.factory()
									 .setTask( this )
									 .build();
			
		}//try
		
		return results;
	
	}//method

		
    public static void shutdownAnExecutor( final ExecutorService threadEx )
    {
        
    	threadEx.shutdown(); // Disable new tasks from being submitted
        
        try {
            
          // Wait a while for existing tasks to terminate
          
          if ( !threadEx.awaitTermination(
                  4, 
                  TimeUnit.SECONDS)) 
          {
              
        	  threadEx.shutdownNow(); // Cancel currently executing tasks
            
          }//if
          
        } catch (InterruptedException ie) {
            
          // (Re-)Cancel if current thread also interrupted
        	threadEx.shutdownNow();
          
          // Preserve interrupt status
          Thread.currentThread().interrupt();
          
        }//try
        
      }//method
    
    
}//class


/**
 * SnappyCompressionBuilder<p>
 * 
 * Fluent API for Path, other properties of SnappyCompression job.<p> 
 */

class SnappyCompressionBuilder
{
	Optional<Path> _path = Optional.absent(); 

	Boolean _validate = false; 

	SnappyCompressionBuilder()
	{
		super();
		
	}//constructor

	
	SnappyCompressionBuilder path(final Path pathIn)
	{
		_path = Optional.of(pathIn);
		
		return this;
		
	}//method

	
	SnappyCompressionBuilder path(final String pathIn)
	{
		_path = Optional.of( Paths.get(pathIn) );
		
		return this;
		
	}//method
	
	
	SnappyCompressionBuilder validation(final Boolean validate)
	{
		_validate = validate;
		
		return this;
		
	}//method

	
	SnappyCompression build()
	{
		if( !_path.isPresent() )
		{
			throw new BackupSettingsException(
					"Path not set for Snappy compression job");
			
		}//if
		
		return new SnappyCompression(this);
		
	}//method
	
	
	CheckedFuture<TaskResults
		<ImmutableMap<String,String>>, BackupException> buildAsFuture()
	{
		return new SnappyCompression(this);
		
	}//method

	
	Callable<TaskResults<ImmutableMap<String,String>>> buildAsCallable()
	{
		return new SnappyCompression(this);

	}//method

	
}//class


/**
 * Number of tasks complete: Guava predicate.<p>
 * 
 * Used to filter collection of SnappyCompression tasks, reporting those  
 * 	have finished.<p>
 * 
 */

class Completed implements Predicate<SnappyCompression>
{
	private Completed()
	{
		super();
		
	}//constructor
	

	@Override
	public boolean apply( final SnappyCompression taskIn ) 
	{
		return taskIn.isDone();
	
	}//method
	
	
	public static Completed factory()
	{
		return new Completed();
		
	}//method

	
}//class


/**
 * Create one SnappyCompression task from one Path.<p>
 * 
 * Path for individual File required.<p>
 * 
 * Guava Function: Path into SnappyCompression. Used to transform List of
 * 	Paths into callables / Checked Futures.<p>
 */

class CompressionTaskCreation implements Function<Path,SnappyCompression>
{
	final Boolean _validateOption = Boolean.valueOf(
			Settings.access()
 					.getProperty("VALIDATION") ); 
	
	
	private CompressionTaskCreation()
	{
		
	}//constructor
	
	
	public static CompressionTaskCreation factory()
	{
		return new CompressionTaskCreation();
		
	}//method
	
	
	@Override
	public SnappyCompression apply( final Path pathIn ) 
	{			
		checkNotNull( pathIn, "null detected path argument" );
		
		return SnappyCompression.builder()
								.path( pathIn )
								.validation( _validateOption )
								.build();
		
	}//method

	
}//class



